package com.matrix.rabbitmqtest.publishandconfirm;

import com.matrix.rabbitmqtest.util.AppKeys;
import com.matrix.rabbitmqtest.util.RabbitMQUtil;
import com.matrix.rabbitmqtest.util.RunningTimer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.stream.IntStream;

/**
 * 异步发布确认
 */
public class AsyncConfirmProducer {

    public static final int MSG_COUNT = 1000;

    public static void main(String[] args) throws IOException {
        // 获取一个信道
        Channel channel = RabbitMQUtil.getChannel();
        // 开启发布确认
        channel.confirmSelect();
        // 声明队列
        channel.queueDeclare(AppKeys.QUEUE_NAME, true, false, true, null);
        // 准备一个跳表来存储发布的消息, 并在确认后将确认了的消息删除, 留下未确认的消息做后续操作.
        ConcurrentSkipListMap<Long, String> skipListMap = new ConcurrentSkipListMap<>();
        // 准备发布确认监听器
        channel.addConfirmListener(
                // 发布成功回调
                (tag, multi) -> {
                    String msg = skipListMap.get(tag);
                    if (multi) {
                        skipListMap.headMap(tag).clear();
                    }
                    skipListMap.remove(tag);
                    System.out.println("发布确认成功, tag: " + tag + ", isMultipleConfirm: " + multi + ", msg: " + msg);
                },
                // 发布失败回调
                (tag, multi) -> {
                    String msg = skipListMap.get(tag);
                    System.err.println("发布确认失败, tag: " + tag + ", isMultipleConfirm: " + multi + ", msg: " + msg);
                });

        // 发送100条消息, 统计耗时
        RunningTimer timer = new RunningTimer();
        IntStream.range(1, MSG_COUNT + 1).forEach(i -> {
            try {
                String msg = "Hello RabbitMQ" + i;
                skipListMap.put(channel.getNextPublishSeqNo(), msg); // 入跳表
                channel.basicPublish("", AppKeys.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes(StandardCharsets.UTF_8)); // 发布
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        System.out.println("发布" + MSG_COUNT + "条消息, 耗时:" + timer.end() + "ms"); // 发布1000条消息, 耗时:296ms
        System.out.println("未确认的消息数: " + skipListMap.size());
    }

}
